import com.zuikaku.pojo.People;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

public class Main {

    public static void main(String[] args) {
        List<People> peopleList = Arrays.asList(new People("张三",18,120.5f),
                new People("李四",19,130.5f),
                new People("王五",20,80.5f));

        List<People> p2 = Arrays.asList(new People("赵六",18,120.5f),
                new People("七七",19,130.5f),
                new People("八八",20,80.5f));
        //jdk8 stream 流实现 求和 平均
        int avg = peopleList.stream().collect(Collectors.averagingInt(People::getAge)).intValue();
        float totalWeight = (float) peopleList.stream().collect(Collectors.summarizingDouble(People::getWeight)).getSum();
        System.out.println("avg:"+avg+" totalWeight:"+totalWeight);

        //Flink元组类
        Tuple3<String,Integer,People> tuple3 = Tuple3.of("chinese",1,new People("zhangsan",12,80.5f));
        System.out.println("first:"+tuple3.f0+" second:"+tuple3.f1+" third:"+tuple3.f2);

        //Flink流处理案例
        //1.取得流运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);
        //2.获得source
        DataStream<String> source = env.fromElements("zhangsan,18,100","lisi,20,120","wangwu,15,60");
        source.print("源数据(source)");
        //3.进行transformation并得到sink
        DataStream<People> sink = source.flatMap(new FlatMapFunction<String, People>() {
            @Override
            public void flatMap(String s, Collector<People> collector) throws Exception {
                String[] temp = s.split(",");
                String name = temp[0];
                int age = Integer.parseInt(temp[1]);
                float weight =Float.parseFloat(temp[2]);
                People p = new People(name,age,weight);
                collector.collect(p);
            }
        });
        sink.print("处理后(sink)");
        //4.开启job(job默认并行执行，且无顺序)
        try {
            env.execute("myJob");
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
